package com.example.kafkademo.transaction;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

/**
 * @author zyy
 * @version 1.0
 * @date 2020/10/20 17:28
 *
 */
public class KafkaTransactionTest {

    public static void main(String[] args){
        Properties properties = new Properties();
        properties.put("bootstrap.servers", "10.42.25.12:9092,10.42.25.13:9092,10.42.25.15:9092,");
        properties.put("batch.size", 65536);
        properties.put("linger.ms", 10);
        properties.put("buffer.memory", 134217728);
        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        // 开启GZIP压缩
        properties.put("compression.type", "zstd");   //Producer 端压缩、Broker 端保持、Consumer 端解压缩。
        properties.put("acks", "all");   //表明所有副本 Broker 都要接收到消息，该消息才算是“已提交”
        // TODO  消息重试 和 callback 有什么区别？     个人理解重试可以处理网络抖动， callback 可以处理其他异常
        properties.put("retries", "3");  // kafka 消息重试次数
        properties.put("enable.idempotence", true);   // 事务消息

        Producer<String, String> producer =  new KafkaProducer<>(properties);
        producer.initTransactions();
        try {
            producer.beginTransaction();
            for (int i = 0; i < 10; i++) {
                int sum = i + 5;
                /**
                 * Kafka 默认分区策略实际上同时实现了两种策略：如果指定了 Key，那么默认实现按消息键保序策略；如果没有指定 Key，则使用轮询策略。
                 */
                producer.send(new ProducerRecord<String, String>("test-a", "1",  String.valueOf(sum)));  //默认轮询，有key hash 发往指定分区
                producer.send(new ProducerRecord<String, String>("test-a", "2",  String.valueOf(sum)));  //默认轮询，有key hash 发往指定分区
                producer.send(new ProducerRecord<String, String>("test-a", "3",  String.valueOf(sum)));  //默认轮询，有key hash 发往指定分区
                producer.send(new ProducerRecord<String, String>("test-a", "4",  String.valueOf(sum)));  //默认轮询，有key hash 发往指定分区
                producer.send(new ProducerRecord<String, String>("test-a", "5",  String.valueOf(sum)));  //默认轮询，有key hash 发往指定分区
                producer.send(new ProducerRecord<String, String>("test-a", "6",  String.valueOf(sum)));  //默认轮询，有key hash 发往指定分区
            }
            producer.commitTransaction();

        } catch (Exception e) {
            producer.abortTransaction();
            e.printStackTrace();
        } finally {
            producer.close();  //KafkaProducer  对象实现了Closeable, try-with-resources 会自动关闭资源
        }

    }



}
